package com.atguigu.Flink.wordCount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

//并行度: 一个Flink程序并行执行的程度。 简单理解将程序复制成多份，同时执行。
//        *
//        * 并行度的设置:
//        *    1. 如果在IDEA中执行，默认的并行度就是CPU最大的线程数。
//        *    2. 在代码中，给每个算子单独设置并行度。
//        *    3. 在代码中，给作业设置全局并行度。
//        *    4. 在提交作业的时候设置并行度(推荐使用)
//        *          bin/flink run -p 2  -d  -c Xxx ./xx.jar
//        *    5. 如果代码中和提交作业时都没有指定过并行度， 在集群中运行时默认的并行度取配置文件配置的并行度。
//        *       parallelism.default: 1
//        *
//        * 并行度设置的优先级:
//        *     代码中单个算子  >  代码中全局  > 提交作业指定  > 配置文件中配置
//        *
//        * 一个作业的并行度如何确定?
//        *    作业的并行度由当前作业中最大并行度的task来决定。
//        *    作业的并行度 = 最大并行度的Task的并行度

public class FLink06_Parallelism {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 5678);
        // 1. 准备流式执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        //设置全局并行度
        env.setParallelism(1);

        // 2. 从数据源读取数据
        // DataStreamSource => DataStream
       // DataStreamSource<String> ds = env.socketTextStream("hadoop102", 8888);
        DataStreamSource<String> ds = env.readTextFile("Input/words.txt");

        // 3. 转换处理
        // 3.1 将读取到的一行数据按照分隔符进行切分，处理成（word, 1）的格式
        // SingleOutputStreamOperator => DataStream
        SingleOutputStreamOperator<Tuple2<String, Long>> flatMap = ds.flatMap(
                new FlatMapFunction<String, Tuple2<String, Long>>() {
                    @Override
                    public void flatMap(String value, Collector<Tuple2<String, Long>> out) throws Exception {
                        String[] words = value.split(" ");
                        for (String word : words) {
                            out.collect(Tuple2.of(word, 1L));
                        }
                    }
                }
        );
        //       .setParallelism(2) ;

        //3.2 根据单词分组
        //KeyedStream => DataStream
        KeyedStream<Tuple2<String, Long>, String> keyBy = flatMap.keyBy(
                new KeySelector<Tuple2<String, Long>, String>() {
                    @Override
                    public String getKey(Tuple2<String, Long> value) throws Exception {
                        return value.f0; // 使用Tuple的第一个元素作为Key
                    }
                }
        );
        //3.3 统计每个单词出现的次数
        /*
           sum(int) : 如果流中的数据类型是Tuple, 指定Tuple中的第几个元素用于sum汇总。
           sum(String) : 如果流中的数据类型是POJO ， 指定POJO中的哪个属性用于sum汇总。
         */
        SingleOutputStreamOperator<Tuple2<String, Long>> sum = keyBy.sum(1).setParallelism(3);

        //4. 输出结果
        sum.print().setParallelism(4);

        //5. 启动执行
        env.execute();
    }
}
